#!/bin/bash
#定义脚本变量
JOB_LIST=(10.10.104.52 10.10.104.56)
JOB_LOCAL_IP=10.10.104.38
TASK_IP=10.10.104.38
LOG_FILE='/tmp/log_flink_status.log'

#加载系统变量，并打印时间
source /etc/profile
# flink TaskManagerRunner运行状态检测模块
check_flink_taskmanager () {
  #判断jobmanager API是否正常
  for job_test_ip in ${JOB_LIST[@]}
  do
    JOB_API_STATUS=$(curl -I -s -X GET http://${job_test_ip}:8081/jobmanager/metrics?get=taskSlotsAvailable | grep "200")
    if [[ ${JOB_API_STATUS} =~ 'OK' ]];then
      JOB_IP=${job_test_ip}
      break
    fi
  done
  #查看flink TaskManagerRunner运行状态
  flink_taskmanager_status=$(jps | grep -w "TaskManagerRunner" | awk {'print $NR'})
  flink_taskmanager_running_num=$(jps | grep -w "TaskManagerRunner" | awk {'print $NR'} | wc -l)
  #判断是否需要重启TaskManagerRunner运行状态
  if [[ ${flink_taskmanager_status:-none} = 'none' && ${flink_taskmanager_running_num} -eq 0 ]];then
    echo 'flink task is down,now start' >> ${LOG_FILE}
    /usr/local/flink-1.17.1/bin/taskmanager.sh start
  elif [[ ${flink_taskmanager_status:-none} != 'none' && ${flink_taskmanager_running_num} -gt 1 ]];then
    echo 'flink task is start too many' >> ${LOG_FILE}
    kill -9 ${flink_taskmanager_status} 
    sleep 5
    /usr/local/flink-1.17.1/bin/taskmanager.sh start
  else
    echo 'flink task is up' >> ${LOG_FILE}
  fi
  #查看flink TaskManagerRunner是否在flink集群内
  flink_taskmanager_heartbeat_status=$(curl -s --location "http://${JOB_IP}:8081/taskmanagers")
  if [[ ${flink_taskmanager_heartbeat_status} =~ "${TASK_IP}" ]];then
    echo "task heartbeat is running" >> ${LOG_FILE}
  else
    echo "task heartbeat is no running,now restart" >> ${LOG_FILE}
    /usr/local/flink-1.17.1/bin/taskmanager.sh stop
    sleep 5
    /usr/local/flink-1.17.1/bin/taskmanager.sh start
  fi
}
# flink jobmanagerRunner运行状态检测模块
check_flink_jobmanager () {
  #查看flink JobManagerRunner运行状态
  flink_jobmanager_status=$(jps | grep -w "StandaloneSessionClusterEntrypoint" | awk {'print $NR'})
  flink_jobmanager_running_num=$(jps | grep -w "StandaloneSessionClusterEntrypoint" | awk {'print $NR'} | wc -l)
  #判断是否需要重启JobManagerRunner运行状态
  if [[ ${flink_jobmanager_status:-none} = 'none' && ${flink_jobmanager_running_num} -eq 0 ]];then
    echo 'flink jobmanager is down,now start' >> ${LOG_FILE}
    /usr/local/flink-1.17.1/bin/jobmanager.sh start
  elif [[ ${flink_jobmanager_status:-none} != 'none' && ${flink_jobmanager_running_num} -gt 1 ]];then
    echo 'flink jobmanager is start too many' >> ${LOG_FILE}
    kill -9 ${flink_jobmanager_status}
    sleep 5
    /usr/local/flink-1.17.1/bin/jobmanager.sh start
  else
    echo 'flink jobmanager is up' >> ${LOG_FILE}
  fi
  #查看flink JobManagerRunner API 是否正常
  JOB_STATUS=$(curl -I -s -X GET http://${JOB_LOCAL_IP}:8081/jobmanager/metrics?get=taskSlotsAvailable | grep "200")  
  if [[ ${JOB_STATUS} =~ 'OK' ]];then
    echo "jobmanager api is ok!" >> ${LOG_FILE}
  else
    echo "jobmanager api is false!,now restart" >> ${LOG_FILE}
    /usr/local/flink-1.17.1/bin/jobmanager.sh stop
    sleep 5
    /usr/local/flink-1.17.1/bin/jobmanager.sh start
  fi
}
case $1 in
  jobmanager)
    echo "$(date +%Y/%m/%d-%T)==========" >> ${LOG_FILE}
    check_flink_jobmanager
  ;;
  taskmanager)
    echo "$(date +%Y/%m/%d-%T)==========" >> ${LOG_FILE}
    check_flink_taskmanager
  ;;
  *)
    echo "pleace run: $0 jobmanager|taskmanager"
  ;;
esac

